package com.shujia.spark.stream

import org.apache.spark.streaming.dstream.{DStream, ReceiverInputDStream}
import org.apache.spark.streaming.{Durations, StreamingContext}
import org.apache.spark.{SparkConf, SparkContext}

object Demo2UpdateStateByKey {
  def main(args: Array[String]): Unit = {
    //创建spark环境
    val conf = new SparkConf()
    conf.setAppName("state")
    conf.setMaster("local[2]")
    val sc = new SparkContext(conf)

    val ssc = new StreamingContext(sc, Durations.seconds(5))

    ssc.checkpoint("data/checkpoint")

    //读取数据
    val linesDS: ReceiverInputDStream[String] = ssc.socketTextStream("master", 8888)

    //转换成kv格式
    val kvDS: DStream[(String, Int)] = linesDS
      .flatMap(line => line.split(","))
      .map(word => (word, 1))


    /**
     * updateFun: 每一个key，每隔批次执行一次
     *
     * @param seq   :当前批次一个keu所有的value
     * @param state :之前批次的计算结果（状态），如果是第一个批次，状态中没有数据
     * @return 返回最新计算结果
     */
    def updateFun(seq: Seq[Int], state: Option[Int]): Option[Int] = {
      println(s"当前批次的数据：$seq")
      println(s"之前批次的状态：$state")
      //1、获取之前的计算结果
      //取出状态中保存的单词的数量，如果没有值，返回0
      val lastCount: Int = state.getOrElse(0)

      //2、计算当前批次单词的数量
      val count: Int = seq.sum

      //3、放回总的单词的数量
      Option(lastCount + count)
    }

    /**
     * updateStateByKey: 有状态算子，每一次计算基于前一次的状态进行累加计算
     * 需要设置checkpoint，用于保存计算的状态
     */
    val countDS: DStream[(String, Int)] = kvDS.updateStateByKey(updateFun)

    countDS.print()


    ssc.start()
    ssc.awaitTermination()
    ssc.stop()

  }

}
